#include "forward_service.h"
#include "forward_util.h"
#include "session_factory.h"
#include "java_call.h"
#include "mutex.h"
#include "session_recycler.h"

#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <stdio.h>
#include <errno.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <assert.h>

#define EPOLL_SIZE 		10000
#define BACKLOG_SIZE	10

/**
* c++函数
*/

forward_service* forward_service::instance = 0;

forward_service* forward_service::get_instance(){
    if (instance == NULL){
        instance = new forward_service;
    }
    return instance;
}

void forward_service::destroy(){
    if (instance != NULL){
        delete instance;
        instance = NULL;
    }
}

forward_service::forward_service(void) :
running(false),
tid(0),
forward_port_tcp(TCP_FORWARD_PORT),
forward_port_udp(UDP_FORWARD_PORT)
{
	session_map.clear();
}

forward_service::~forward_service(void){
}

bool forward_service::is_running(){
    return running;
}

bool forward_service::start_service(int port){

    bool ret = false;

	assert(0);

    forward_port_tcp = port;

    if (!is_running()){
        int result = pthread_create(&tid, NULL, work_thread, this);
        if (result == 0){
            ret = true;
        }
    }else{
        ret = true;
    }

    return ret;
}

bool forward_service::stop_service(){
    bool ret = true;
    if (is_running()){
        running = false;
        if (tid > 0){
            pthread_join(tid, NULL);
        }
    }
    return ret;
}

void forward_service::work_thread_impl(){

	int ret = -1;
	forward_event tcp_srv_evt, udp_srv_evt;

    socklen_t socklen = sizeof(struct sockaddr_in);

    ep_monitor.init();

	do {
		/**
		 * step 1. tcp listen
		 */
		LOGI("create tcp listen socket ...");
		tcp_srv.auto_release(true);
		ret = tcp_srv.create(forward_socket::TCP);
		if (ret == -1) {
			break;
		}
		ret = tcp_srv.nonblocking(true);
		if (ret == -1) {
			break;
		}
		ret = tcp_srv.reuseaddr(true);
		if (ret == -1) {
			break;
		}
		ret = tcp_srv.bind(forward_port_tcp);
		if (ret == -1) {
			LOGE("bind failed, %d", errno);
			break;
		}
		ret = tcp_srv.listen(BACKLOG_SIZE);
		if (ret == -1) {
			LOGE("tcp_srv listen failed, %d", errno);
			break;
		}
		tcp_srv_evt.role = forward_event::LISTENER;
		tcp_srv_evt.protocol = forward_event::TCP;
		tcp_srv_evt.fd = tcp_srv.fd();
		ret = ep_monitor.monitor_add(tcp_srv.fd(), &tcp_srv_evt,
				EPOLLIN | EPOLLET);
		if (ret < 0) {
			LOGE("monitor_add tcp_srv failed! %d", errno);
			break;
		}

		/**
		 * step 2. udp listen
		 */
		LOGI("create udp listen socket ...");
		udp_srv.auto_release(true);
		ret = udp_srv.create(forward_socket::UDP);
		LOGE("udp_srv = %d", ret);
		if (ret == -1) {
			break;
		}
		ret = udp_srv.nonblocking(true);
		if (ret == -1) {
			break;
		}
		ret = udp_srv.reuseaddr(true);
		if (ret == -1) {
			break;
		}
		ret = udp_srv.bind(forward_port_udp);
		if (ret < 0) {
			LOGE("udp_srv bind failed!");
			break;
		}
		udp_srv_evt.role = forward_event::IN;
		udp_srv_evt.protocol = forward_event::UDP;
		udp_srv_evt.fd = udp_srv.fd();
		ret = ep_monitor.monitor_add(udp_srv.fd(), &udp_srv_evt,
				EPOLLIN | EPOLLOUT | EPOLLET);
		if (ret < 0) {
			LOGE("monitor_add udp_srv failed! %d", errno);
			break;
		}

		LOGI("forward service enter poll ...");

		running = true;
		while (is_running()) {

			int epoll_events_count = ep_monitor.wait(1000);

			for (int i = 0; i < epoll_events_count; i++) {
				// 获取转发事件指针
				const forward_event* cur_fwd_evt =
						(forward_event*) (ep_monitor.events[i].data.ptr);
				if (cur_fwd_evt == NULL) {
					continue;
				}

				// IO事件
				int cur_io_evt = ep_monitor.events[i].events;

				const string role_str[] = { "idle", "in", "out", "listener",
						"max" };
				const string protocol_str[] = { "idle", "tcp", "udp", "max" };
				LOGI("[epoll] io = %d, protocol = %s, role = %s", cur_io_evt, protocol_str[cur_fwd_evt->protocol].c_str(), role_str[cur_fwd_evt->role].c_str());

				if (cur_fwd_evt->protocol == forward_event::TCP) {
					handle_tcp(cur_fwd_evt, cur_io_evt);
				} else if (cur_fwd_evt->protocol == forward_event::UDP) {
					handle_udp(cur_fwd_evt, cur_io_evt);
				}
			}

			// 回收会话
			recycle();			
		}
	} while (0);

	ep_monitor.release();
	tcp_srv.close();
	udp_srv.close();
}

void* forward_service::work_thread(void* param){

    forward_service* self = (forward_service*)param;

    LOGI("forward service started~~~");

    self->work_thread_impl();

    LOGE("forward service ended, errno=%d ~~~", errno);
}

forward_session* forward_service::get_session(unsigned short src_port){
	forward_session* ret = NULL;
	session_map_iterator it = session_map.find(src_port);
	if(it != session_map.end()){
		ret = it->second;
	}
	LOGI("get_session port = %u, ret = %u", src_port, (unsigned int)ret);
	return ret;
}

void forward_service::add_session(forward_session* sess){
	if (!sess){
		return;
	}

	forward_pair& new_fwd_pair = sess->get_pair();
	unsigned short src_port = new_fwd_pair.get_src_port();

	forward_session* exists = get_session(src_port);
	LOGI("add_session %s exists = %u", new_fwd_pair.to_string().c_str(), (unsigned int)exists);
	if (exists != NULL){
		sess->set_pair(new_fwd_pair);
	}else{
		std::pair<unsigned short, forward_session*> new_sess_pair;
		new_sess_pair.first = src_port;
		new_sess_pair.second = sess;
		session_map.insert(new_sess_pair);
	}
}

void forward_service::remove_session(unsigned short src_port){
	
	forward_session* ptr = get_session(src_port);
	LOGI("remove_session %u, ptr = %u", src_port, (unsigned int)ptr);
	if (ptr){
		session_map.erase(src_port);
		delete ptr;
	}
}

forward_session* forward_service::safe_get_session(unsigned short src_port){

	forward_session* ret = NULL;

	struct timeval begin, end;
	gettimeofday(&begin, NULL);

	pair_map_mutex.lock();

	session_map_iterator it = session_map.find(src_port);
	if(it != session_map.end()){
		ret = it->second;
	}

	pair_map_mutex.unlock();

	//print_pair_map();
	gettimeofday(&end, NULL);
	LOGI("safe_get_session port = %u, interval = %lu us", src_port , end.tv_usec - begin.tv_usec);

	return ret;
}

void forward_service::safe_add_session(forward_session* sess){
	pair_map_mutex.lock();
	add_session(sess);
	pair_map_mutex.unlock();
}

void forward_service::safe_remove_session(unsigned short src_port){
	pair_map_mutex.lock();
	remove_session(src_port);
	pair_map_mutex.unlock();
}

forward_pair forward_service::safe_get_session_pair(unsigned short src_port){

	bool ret = false;
	forward_pair fwd_pair;
	
	struct timeval begin, end;
	gettimeofday(&begin, NULL);

	pair_map_mutex.lock();

	session_map_iterator it = session_map.find(src_port);
	if(it != session_map.end()){
		forward_session* sess = it->second;
		fwd_pair = sess->get_pair();
		ret = true;
	}
	pair_map_mutex.unlock();

	gettimeofday(&end, NULL);
	LOGI("safe_get_session_pair port = %u, ret= %d, interval = %lu us", src_port , ret, end.tv_usec - begin.tv_usec);
	return fwd_pair;
}

void forward_service::print_pair_map(){
	forward_pair* ret = NULL;

	LOGI("================ print_pair_map ================");
	int i = 1;
	for(session_map_iterator it = session_map.begin();it!=session_map.end();++it,++i){
		forward_session* fs = it->second;
		LOGI("%d %s", i, fs->get_pair().to_string().c_str());
	}
	LOGI("-- end --");
}

void forward_service::handle_tcp(const forward_event* cur_fwd_evt, int cur_io_evt){

	// 函数返回值
	int ret = 0;
	// 当前转发会话
	forward_session_tcp* fwd_session = NULL;

	do{
		if(cur_fwd_evt->role == forward_event::LISTENER){
			if (cur_io_evt & EPOLLIN){
				LOGI("EPOLLIN");

				// 当前IO事件的fd
				int cur_fd = cur_fwd_evt->fd;
				handle_accept(cur_fd);
			}
			if (cur_io_evt & EPOLLOUT){
				LOGI("EPOLLOUT");
			}
			if (cur_io_evt & EPOLLERR){
				LOGI("EPOLLERR");
			}
		}
		else if (cur_fwd_evt->role == forward_event::IN)
		{
			fwd_session = (forward_session_tcp*)cur_fwd_evt->ptr_session;
			if (fwd_session == NULL){
				break;
			}

			LOGI("%s", fwd_session->to_string());

			if (cur_io_evt & EPOLLIN){
				LOGI("EPOLLIN");
				ret = fwd_session->packet_forward(forward_session::OUTWARD);
				if (ret == -1){
					break;
				}
			}
			if (cur_io_evt & EPOLLOUT){
				LOGI("EPOLLOUT");
				// 尽力发出挂起包
				ret = fwd_session->pend_forward(forward_session::INWARD);
				if (ret == -1){
					break;
				}else if (ret == 0){
					// 没有挂起包，没东西可发
				}else if (ret == 1){
					// 下次再说
				}else if (ret == 2){
					// 看看还有没有要发的
					ret = fwd_session->packet_forward(forward_session::INWARD);
					if (ret == -1){
						break;
					}
				}
			}
			if (cur_io_evt & EPOLLERR){
				LOGI("EPOLLERR");
				ret = -1;
				break;
			}
		}
		else if (cur_fwd_evt->role == forward_event::OUT)
		{
			fwd_session = (forward_session_tcp*)cur_fwd_evt->ptr_session;
			if (fwd_session == NULL){
				break;
			}

			LOGI("%s", fwd_session->to_string());

			if (cur_io_evt & EPOLLIN){
				LOGI("EPOLLIN");
				ret = fwd_session->packet_forward(forward_session::INWARD);
				if (ret == -1){
					break;;
				}
			}
			if (cur_io_evt & EPOLLOUT){
				LOGI("EPOLLOUT");
				// 如果正在连接out端，判断为连接成功
				if (fwd_session->get_status() != forward_session::CONNECTED){
					fwd_session->set_status(forward_session::CONNECTED);

					// 开始监听入口事件
					forward_event& evt_in = fwd_session->get_evt_in();
					evt_in.role = forward_event::IN;
					evt_in.protocol = forward_event::TCP;
					evt_in.ptr_session = fwd_session;
					ret = ep_monitor.monitor_add(fwd_session->get_fd_in(), &evt_in, EPOLLIN | EPOLLOUT | EPOLLET);
					if (ret < 0){
						LOGE("monitor_add IN failed!");
						ret = -1;
						break;
					}
				}

				// 尽力发出挂起包
				ret = fwd_session->pend_forward(forward_session::OUTWARD);
				if (ret == -1){
					break;
				}else if (ret == 0){
					// 没有挂起包，没东西可发
				}else if (ret == 1){
					// 下次再说
				}else if (ret == 2){
					// 看看还有没有要发的
					ret = fwd_session->packet_forward(forward_session::OUTWARD);
					if (ret == -1){
						break;
					}
				}
			}
			if (cur_io_evt & EPOLLERR){
				LOGI("EPOLLERR");
				ret = -1;
				break;
			}
		}
	}while(0);

	// 清理工作
	if (ret == -1){
		if (fwd_session != NULL){
			delete fwd_session;
			fwd_session = NULL;
		}
	}		
}

void forward_service::handle_udp(const forward_event* cur_fwd_evt, int cur_io_evt){

	int ret = 0;

	do{
		char recv_buffer[256] = {0};
		const size_t recv_buf_len = sizeof(recv_buffer);
		int recv_len = 0;
		struct sockaddr_in from;
		socklen_t sockaddr_len = sizeof(struct sockaddr);

		if(cur_fwd_evt->role == forward_event::IN){
			if (cur_io_evt & EPOLLIN){
				LOGI("EPOLLIN");

				while(true){
					recv_len = udp_srv.recvfrom(recv_buffer, recv_buf_len, 0, (struct sockaddr*)&from , &sockaddr_len);
					LOGI("recvfrom %s ret=%d, errno=%d", udp_srv.to_string(), recv_len, errno);
					if (recv_len == -1){
						if (errno == EAGAIN || errno == EWOULDBLOCK){
							break;
						}else{
							break;
						}
					}

					// 对端地址
					string from_addr_str = forward_util::get_addr_by_sockaddr_in(from);
					unsigned short peer_port = forward_util::get_port_by_sockaddr_in(from);
					LOGI("from = %s:%u", from_addr_str.c_str(), peer_port);
					forward_session_udp* sess = (forward_session_udp*)safe_get_session(peer_port);
					if (sess == NULL){
						break;
					}

					sockaddr_in& to = sess->get_pair().si_out;
					string to_addr_str = forward_util::get_sockaddr_str(to);
					LOGI("to = %s", to_addr_str.c_str());

					// 如果没有创建outgoing socket则创建并绑定到会话
					forward_socket fd_out;
					fd_out.attach(sess->get_fd_out());
					if (!fd_out.is_valid()){
						ret = fd_out.create(forward_socket::UDP);
						LOGE("fd_out = %d", ret);
						if (ret == -1){
							break;
						}
						ret = fd_out.nonblocking(true);
						if (ret == -1){
							break;
						}						
						sess->set_fd_out(fd_out.fd());

						// 保护这个socket
						call_java_protectSocket(fd_out.fd());

						// 开始监听出口事件
						forward_event& evt_out = sess->get_evt_out();
						evt_out.role = forward_event::OUT;
						evt_out.protocol = forward_event::UDP;
						evt_out.ptr_session = sess;
						ret = ep_monitor.monitor_add(sess->get_fd_out(), &evt_out, EPOLLIN | EPOLLOUT | EPOLLET);
						if (ret < 0){
							LOGE("monitor_add OUT failed!");
							ret = -1;
							break;
						}
					}

					// 重置会话时间
					session_recycler::get_instance()->reorder(sess->td);
					
					int send_len = fd_out.sendto(recv_buffer, recv_len, 0, (struct sockaddr*)&to, sockaddr_len);
					LOGI("sendto %s ret=%d, errno=%d", fd_out.to_string(), recv_len, errno);
					if (send_len == -1){
						if (errno == EAGAIN || errno == EWOULDBLOCK){
							break;
						}else{
							break;
						}
					}
				}
			}
			if (cur_io_evt & EPOLLOUT){
				LOGI("EPOLLOUT");


			}
			if (cur_io_evt & EPOLLERR){
				LOGI("EPOLLERR");
			}
		}
		else if (cur_fwd_evt->role == forward_event::OUT){

			forward_session_udp* sess = (forward_session_udp*)cur_fwd_evt->ptr_session;
			if (sess == NULL){
				break;
			}

			if (cur_io_evt & EPOLLIN){
				LOGI("EPOLLIN");

				while(true){
					forward_socket fd_out;
					fd_out.attach(sess->get_fd_out());
					if (!fd_out.is_valid()){
						break;
					}
					
					recv_len = fd_out.recvfrom(recv_buffer, recv_buf_len, 0, (struct sockaddr*)&from , &sockaddr_len);
					LOGI("recvfrom %s ret=%d, errno=%d", fd_out.to_string(), recv_len, errno);
					if (recv_len == -1){
						if (errno == EAGAIN || errno == EWOULDBLOCK){
							break;
						}else{
							break;
						}
					}

					// 重置会话时间
					session_recycler::get_instance()->reorder(sess->td);

					sockaddr_in si_in = sess->get_pair().si_in;
					si_in.sin_addr.s_addr = inet_addr(STUB_IP);
					string to_str = forward_util::get_sockaddr_str(si_in);
					int send_len = udp_srv.sendto(recv_buffer, recv_len, 0, (struct sockaddr*)&si_in, sockaddr_len);
					LOGI("sendto %s %s ret=%d, errno=%d", udp_srv.to_string(), to_str.c_str(), recv_len, errno);
					if (recv_len == -1){
						if (errno == EAGAIN || errno == EWOULDBLOCK){
							break;
						}else{
							break;
						}
					}
				}
			}

			if (cur_io_evt & EPOLLOUT){
				LOGI("EPOLLOUT");


			}
			if (cur_io_evt & EPOLLERR){
				LOGI("EPOLLERR");
			}
		}

	}while(0);
}

void forward_service::handle_accept(int cur_fd){

	socklen_t socklen = sizeof(struct sockaddr_in);

	while (true){

		int ret = 0;
		forward_socket entrance, exit, listener;
		int accept_socket = -1;
		forward_session_tcp* fwd_session = NULL;

		listener.attach(cur_fd);

		do{
			struct sockaddr_in their_addr;
			accept_socket = listener.accept((struct sockaddr *) &their_addr, &socklen);
			LOGI("accept ret=%d, errno=%d", accept_socket, errno);
			if (accept_socket < 0){
				LOGE("accept failed, %d", errno);
				ret = -1;
				break;
			}

			entrance.attach(accept_socket);
			ret = entrance.nonblocking(true);
			if (ret < 0){
				LOGE("setnonblocking failed!");
				ret = -1;
				break;
			}

			// 对端地址
			string peer_addr = forward_util::get_addr_by_sockaddr_in(their_addr);
			unsigned short peer_port = forward_util::get_port_by_sockaddr_in(their_addr);
			LOGI("peer =%s:%u", peer_addr.c_str(), peer_port);
			fwd_session = (forward_session_tcp*)safe_get_session(peer_port);
			if (fwd_session == NULL){
				ret = -1;
				break;
			}

			// 创建出口fd并绑定到会话
			ret = exit.create(forward_socket::TCP);
			if (ret < 0){
				LOGE("create exit socket failed!");
				ret = -1;
				break;
			}
			ret = exit.nonblocking(true);
			if (ret < 0){
				LOGE("exit setnonblocking failed!");
				ret = -1;
				break;
			}
			call_java_protectSocket(exit.fd());

			fwd_session->attach_fd_pair(entrance.fd(), exit.fd());
			
			const sockaddr_in& cnn_addr = fwd_session->get_pair().si_out;
			ret = connect(exit.fd(), (sockaddr*)&cnn_addr, sizeof(cnn_addr));
			LOGI("connect ret %d, errno %d", ret, errno);
			if (ret == -1){
				if (errno != EINPROGRESS){
					ret = -1;
					break;
				}
			}

			forward_event& evt_out = fwd_session->get_evt_out();
			evt_out.role = forward_event::OUT;
			evt_out.protocol = forward_event::TCP;
			evt_out.ptr_session = fwd_session;

			int fd_out = fwd_session->get_fd_out();
			ret = ep_monitor.monitor_add(fd_out, &evt_out, EPOLLIN | EPOLLOUT | EPOLLET);
			if (ret < 0){
				LOGE("monitor_add OUT failed!");
				ret = -1;
				break;
			}

		}while(0);

		// 清理工作
		if (ret == -1){
			entrance.close();
			exit.close();			
			if (fwd_session != NULL){
				fwd_session->reset();
				delete fwd_session;
				fwd_session = NULL;
			}
			break;
		}
	};
}

void forward_service::recycle(){
	session_recycler* rec = session_recycler::get_instance();
	
	while(true){
		timer_domain* domain = rec->front();
		if (domain == NULL){
			break;
		}
		forward_session* sess = domain->session_ptr;
		if (sess == NULL){
			// 异常
			break;
		}

		if (domain->expire()){
			unsigned short port = sess->get_pair().get_src_port();
			safe_remove_session(port);
		}
		break;		
	}
}

